package com.atguigu.edu.realtime.dwd.db.app;

import com.atguigu.edu.realtime.common.base.BaseSQLApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Examine_Answer extends BaseSQLApp {
    public static void main(String[] args) {
        new Examine_Answer().start(
                1256,
                4
                , Constant.TOPIC_DB
        );
    }

    @Override
    public void handle1(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        readOdsDS(tableEnv,"dwd_exam_question");

        Table examTable = tableEnv.sqlQuery(
                "select " +
                        " `data`['id'] exam_id ," +
                        " `data`['paper_id'] paper_id ," +
                        " `data`['user_id'] user_id ," +
                        " ts " +
                        " from topic_db where `table`='test_exam' and `data`['deleted'] = '0' " +
                        " and (`type`='insert' or `type`='update')"
        );
        //examTable.execute().print();
        tableEnv.createTemporaryView("exam_table",examTable);

        Table examQuestionTable = tableEnv.sqlQuery(
                "select " +
                        " `data`['exam_id'] exam_id ," +
                        " `data`['paper_id'] paper_id ," +
                        " `data`['question_id'] question_id ," +
                        " `data`['user_id'] user_id ," +
                        " `data`['is_correct'] is_correct " +
                        " from topic_db where `table`='test_exam_question' and `data`['deleted'] = '0' " +
                        " and (`type`='insert' or `type`='update')"
        );
        //examQuestionTable.execute().print();
        tableEnv.createTemporaryView("test_exam_question_table",examQuestionTable);

        Table dwdExamQuestion = tableEnv.sqlQuery(
                "select et.exam_id , et.paper_id , et.user_id , eq.question_id , eq.is_correct , ts" +
                        " from exam_table as et join test_exam_question_table as eq " +
                        " on et.exam_id = eq.exam_id and et.paper_id = eq.paper_id and et.user_id = eq.user_id "
        );

        //dwdExamQuestion.execute().print();

        tableEnv.executeSql(
                "create table dwd_exam_question ( " +
                        " exam_id string ," +
                        " paper_id string ," +
                        " user_id string , " +
                        " question_id string ," +
                        " is_correct string , " +
                        " ts bigint ," +
                        " PRIMARY KEY (exam_id,paper_id,user_id) NOT ENFORCED" +
                        " ) "+ SQLUtil.getUpsertKafkaDDL("dwd_exam_question")
        );

        dwdExamQuestion.executeInsert("dwd_exam_question");
    }
}
